package main

import (
	"context"
	"github.com/segmentio/kafka-go"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"
)

func main() {
	// 1. 配置参数（根据实际环境修改）
	brokers := []string{"localhost:9091"} // Kafka broker地址
	topic := "first_topic"                // 目标主题
	groupID := "group_consumer_1"         // 消费者组ID（取消注释并设置正确的组ID）

	// 2. 配置消费者组模式的Reader
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  brokers,
		GroupID:  groupID, // 指定消费者组ID
		Topic:    topic,
		MinBytes: 1,    // 最小拉取字节数
		MaxBytes: 10e6, // 最大拉取字节数（10MB）

		// 消费者组模式下，这些参数由组协调器控制，通常不需要设置
		// MaxWait:        100 * time.Millisecond, // 注释掉，避免与组协调器冲突
		// CommitInterval: 1 * time.Second,        // 注释掉，使用自动提交或手动提交

		// 可选优化配置
		ReadBackoffMin: 100 * time.Millisecond,
		ReadBackoffMax: 500 * time.Millisecond,
		Dialer: &kafka.Dialer{
			Timeout:   5 * time.Second,
			KeepAlive: 30 * time.Second,
		},
	})

	// 3. 启动消费循环
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			default:
				// 读取单条消息
				msg, err := r.ReadMessage(ctx)
				if err != nil {
					log.Printf("消费错误: %v（将重试）", err)
					time.Sleep(200 * time.Millisecond)
					continue
				}
				_ = r.CommitMessages(ctx, msg)
				// 打印消息详情
				log.Printf("[实时消息] 主题=%s | 分区=%d | 偏移量=%d | 键=%s | 值=%s",
					msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
			}
		}
	}()

	// 4. 优雅关闭
	waitForShutdown(r, cancel)
}

// 处理程序退出逻辑
func waitForShutdown(r *kafka.Reader, cancel context.CancelFunc) {
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	<-sigChan

	log.Println("正在停止消费者...")
	cancel()
	r.Close()
	log.Println("消费者已安全退出")
}
